package com.eshipenet.shipnet.equipmentextbuilder

import org.apache.flink.streaming.api.scala._
import org.joda.time.DateTime
import play.api.libs.json.{JsObject, JsValue}

object Main {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.getConfig.registerTypeWithKryoSerializer(classOf[DateTime], classOf[DateTimeSerializer])

    val equipmentJsValueStream = createEquipmentJsValueStream(env)
    val equipmentActionStream = equipmentJsValueStream
      .map(jsValue => DebeziumParser(jsValue))
      .filter(_.isSuccess)
      .map(_.get)
      .filter(_.isDefined)
      .map(_.get)
      .map(EquipmentAction(_))
      .filter(_.isSuccess)
      .map(_.get)
    equipmentActionStream.print("equipmentActionStream")

    val equipmentModelLinkJsValueStream = createEquipmentModelLinkJsValueStream(env)
    val equipmentModelLinkActionStream = equipmentModelLinkJsValueStream
      .map(DebeziumParser(_))
      .filter(_.isSuccess)
      .map(_.get)
      .filter(_.isDefined)
      .map(_.get)
      .map(EquipmentModelLinkAction(_))
      .filter(_.isSuccess)
      .map(_.get)
//    equipmentModelLinkActionStream.print("equipmentModelLinkActionStream")

    val modelJsValueStream = createEquipmentModelStream(env)
    val modelActionStream = modelJsValueStream
      .map(DebeziumParser(_))
      .filter(_.isSuccess)
      .map(_.get)
      .filter(_.isDefined)
      .map(_.get)
      .map(EquipmentModelAction(_))
      .filter(_.isSuccess)
      .map(_.get)
    modelActionStream.print("modelUpdateStream")

    val equipmentWithModelIdActionStream =
      Processor.generateEquipmentWithModelIdStream(equipmentActionStream, equipmentModelLinkActionStream)
    equipmentWithModelIdActionStream.print("equipmentWithModelIdActionStream")
    val equipmentWithModeActionStream =
      Processor.generateEquipmentWithModelActionStream(equipmentWithModelIdActionStream, modelActionStream)
    equipmentWithModeActionStream.print("equipmentWithModelActionStream")
    equipmentWithModeActionStream
      .flatMap(EquipmentWithModelActionSerializer(_))
      .addSink(EquipmentWithModelActionStreamSinkProvider.kafkaSink)

    val equipmentExtActionStream =
      Processor.mergeToEquipmentExtAction(equipmentWithModeActionStream)
    equipmentExtActionStream.print("equipmentExtActionStream")
    equipmentExtActionStream
      .flatMap(EquipmentExtActionSerializer(_))
      .addSink(EquipmentExtActionSinkProvider.kafkaSink)

    val name = this.getClass.getName
    env.execute(name)
  }

  private def createEquipmentJsValueStream(env: StreamExecutionEnvironment): DataStream[JsObject] =
    EquipmentActionJsonStreamProvider.kafkaSource(env)

  private def createEquipmentModelLinkJsValueStream(env: StreamExecutionEnvironment): DataStream[JsValue] =
    EquipmentLinkActionJsonStreamProvider.kafkaSource(env)

  private def createEquipmentModelStream(env: StreamExecutionEnvironment): DataStream[JsObject] =
    EquipmentModelActionJsonStreamProvider.kafkaSource(env)

}
